package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.text.SimpleDateFormat;
import java.util.logging.SimpleFormatter;

/**
 * @author jason
 * @create 2022-04-06 20:36
 */

//数据流 web/app -> nginx -> SprinBAoot -> mysql -> flinkapp -> kafka(ods) ->flinkapp -> kafka(dwd)->FlinkApp->Kafka(dwm)
//程序              mockDB   -> MySQL-》 FlinkCDC -> kafka (ZK) -> BaseApp -> Kafka/Phoenix（HBASE，zk，hdfs）


public class UniqueVisitApp {
    public static void main(String[] args) throws Exception {
//        1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

//        //2.Flink-CDC将读取binlog的位置信息以状态的方式保存在CK,如果想要做到断点续传,需要从Checkpoint或者Savepoint启动程序
//        //2.1 开启Checkpoint,每隔5秒钟做一次CK
//        env.enableCheckpointing(5000L);
////        设置超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(10000L);
////        设置最大当前ck数量
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
////        设置最小ck间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);
//        //2.2 指定CK的一致性语义
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        //2.3 设置任务关闭的时候保留最后一次CK数据
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        //2.4 指定从CK自动重启策略
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000L));
//        //2.5 设置状态后端
//        env.setStateBackend(new FsStateBackend("hdfs://djcluster/flinkCDC"));
//        //2.6 设置访问HDFS的用户名
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

//       2. 读取kafka dwd_page_log 主题的数据
        String groupId = "unique_visit_app";
        String sourceTopic = "dwd_page_log";
        String sinkTopic = "dwm_unique_visit";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(sourceTopic, groupId));

        //        3.将每行数据转换为JSON数据
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject);

//        过滤数据 状态编程 只保留每个mid每天第一次登陆的数据
        KeyedStream<JSONObject, String> keyedStream = jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));
        SingleOutputStreamOperator<JSONObject> uvDS = keyedStream.filter(new RichFilterFunction<JSONObject>() {
            private ValueState<String> dataState;
            private SimpleDateFormat simpleDateFormat;

            @Override
            public void open(Configuration parameters) throws Exception {
//状态描述器
                ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("data-state", String.class);
//               给状态设置超时时间
//
                StateTtlConfig stateTtlConfig = new StateTtlConfig
                        .Builder(Time.hours(24))
//                        2022-04-01 09:00:00   2022-04-02 08:00:00  2022-04-02 09:00:00   2022-04-02 10:00:00 状态清除BUG
//                        当前如果重新写入的时候需要重置  比自己写定时器方便
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                        .build();
                valueStateDescriptor.enableTimeToLive(stateTtlConfig);

                dataState = getRuntimeContext().getState(valueStateDescriptor);

                simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");
            }

            @Override
            public boolean filter(JSONObject value) throws Exception {
//               取出上一条页面信息
                String lastPageId = value.getJSONObject("page").getString("last_page_id");
//                判断上一条页面是否为NULL
                if (lastPageId == null || lastPageId.length() <= 0) {
//                    取出状态数据
                    String lastData = dataState.value();

//                    取出今天的日期
                    String curDate = simpleDateFormat.format(value.getLong("ts"));

//                    判断两个日期是否相同  把当前日期放在前面 不会有空指针异常
                    if (!curDate.equals(lastData)) {
                        dataState.update(curDate);
                        return true;
                    }
//                    else {
////                        说明来过
//                        return false;
//                    }
                }
//                else {
                return false;
//                }
            }
        });

//        将数据写入kafka
        uvDS.print();
        uvDS.map(JSONAware::toJSONString).addSink(MyKafkaUtil.getKafkaProducer(sinkTopic));

//        启动任务
        env.execute("UniqueVisitApp");
    }
}
